package xin.alum.aim.cluster;

import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;
import lombok.SneakyThrows;
import org.springframework.scheduling.annotation.Async;
import xin.alum.aim.constant.AIMConstant;
import xin.alum.aim.constant.Constants;
import xin.alum.aim.groups.ClusterFactory;
import xin.alum.aim.groups.Session;
import xin.alum.aim.model.Pusher;
import xin.alum.aim.model.Transportable;

import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

public abstract class ClusterPusher implements ClusterFactory {

    protected final InternalLogger logger = InternalLoggerFactory.getInstance(this.getClass());

    protected abstract <T> boolean push(String channel, T d);

    private Comparator<Pusher<Transportable>> comparator = (p1, p2) -> p1.getData().getPriority() > p2.getData().getPriority() ? p1.getData().getPriority() : p2.getData().getPriority();

    private PriorityBlockingQueue<Pusher<Transportable>> queue = new PriorityBlockingQueue(ClusterFactory.QUEUE_CAPACITY, comparator);


    @Override
    public void kick(Session session) {
        push(Constants.BIND_MESSAGE_INNER_QUEUE, session);
    }

    @Override
    public void push(String recipient, Transportable data) {
        Pusher p = new Pusher(recipient, data);
        queue.put(p);
        push();
    }


    @SneakyThrows
    @Async
    public void push() {
        if (queue.size() > 10) logger.warn("集群消息出现积压:{}/{}", queue.size(), ClusterFactory.QUEUE_CAPACITY);
        while (queue.iterator().hasNext()) {
            Pusher m = queue.poll();
            if (m == null) break;
            switch (m.getData().getType()) {
                case AIMConstant.DATA_TYPE_MESSAGE:
                    push(Constants.PUSH_MESSAGE_INNER_QUEUE, m);
                    break;
                default:
                    push(Constants.REPLY_MESSAGE_INNER_QUEUE, m);
                    break;
            }
        }
    }
}
